//package com.ruoyi.system.chat;
//
//import com.alibaba.fastjson2.JSON;
//import com.ruoyi.common.constant.CacheConstants;
//import com.ruoyi.common.core.redis.RedisCache;
//import com.ruoyi.common.enums.message.MessageIsRead;
//import com.ruoyi.common.enums.session.ConversationType;
//import com.ruoyi.common.utils.SecurityUtils;
//import com.ruoyi.common.utils.StringUtils;
//import com.ruoyi.common.utils.ip.IpUtils;
//import com.ruoyi.common.utils.spring.SpringUtils;
//import com.ruoyi.system.config.ConfigReader;
//import com.ruoyi.system.domain.SysUserOnline;
//import com.ruoyi.system.domain.vo.MessageVO;
//import com.ruoyi.system.domain.vo.SysUserVO;
//import com.ruoyi.system.service.IMessageService;
//import com.ruoyi.system.service.ISysUserOnlineService;
//import lombok.AllArgsConstructor;
//import lombok.Data;
//import lombok.NoArgsConstructor;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.ObjectUtils;
//import org.springframework.beans.BeanUtils;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.util.CollectionUtils;
//
//import java.io.BufferedReader;
//import java.io.IOException;
//import java.io.InputStreamReader;
//import java.io.PrintWriter;
//import java.net.ServerSocket;
//import java.net.Socket;
//import java.util.*;
//import java.util.concurrent.*;
//import java.util.concurrent.atomic.AtomicInteger;
//import java.util.stream.Collectors;
//
//@Slf4j
//public class ChatServer {
//
//
//    private RedisCache redisCache = SpringUtils.getBean(RedisCache.class);
//
//    @Autowired
//    private IMessageService messageService;
//
//    /**
//     * 消息保存线程池(CPU IO密集型任务) - 用于异步保存消息，当用户下线后，触发
//     */
//    private static final ExecutorService saveMsgExecutor = Executors.newFixedThreadPool(
//            Runtime.getRuntime().availableProcessors(), // 获取系统CPU核数
//            new ThreadFactory() {
//                private final AtomicInteger conter = new AtomicInteger(1);
//                @Override
//                public Thread newThread(Runnable r) {
//                    return new Thread(r, "saveMsg-thread-" + conter.getAndIncrement());
//                }
//            }
//    );
//
//    /**
//     * 消息发送线程池(CPU 密集型任务) - 用于异步发送消息
//     */
//    private static final ExecutorService messageSenderExecutor = Executors.newFixedThreadPool(
//            Runtime.getRuntime().availableProcessors(), // 获取系统CPU核数
//            new ThreadFactory() {
//                private final AtomicInteger conter = new AtomicInteger(1);
//                @Override
//                public Thread newThread(Runnable r) {
//                    return new Thread(r, "messageSender-thread-" + conter.getAndIncrement());
//                }
//            }
//    );
//
//    /**
//     * 处理客户端连接线程池(IO密集型任务) - 用于处理客户端的连接，消息转发等
//     */
//    private static final ExecutorService clientHandlerExecutor = Executors.newFixedThreadPool(
//            Runtime.getRuntime().availableProcessors() * 2,
//            new ThreadFactory() {
//                private final AtomicInteger counter = new AtomicInteger(1);
//                @Override
//                public Thread newThread(Runnable r) {
//                    return new Thread(r,"client-handler-"+counter.getAndIncrement());
//                }
//            }
//    );
//
//    /**
//     * 服务器端口
//     */
//    private final int port;
//    /**
//     * 服务地址
//     */
//    private final String host;
//    /**
//     * 聊天用户列表，key为用户id_用户名，value为Socket
//     */
//    private final ConcurrentHashMap<Long, Socket> chatUserList;
//    /**
//     * 用户当前发送的消息列表
//     */
//    private final ConcurrentHashMap<Long, CopyOnWriteArrayList<MessageVO>> userCurMessageList;
//    /**
//     * 用户（RATE_LIMIT_WINDOW_MS）秒内发送的消息队列
//     */
//    private final ConcurrentHashMap<Long, BlockingQueue<MessageVO>> perSecMsgQueue;
//    /**
//     * 用户最后一次发送消息的时间
//     */
//    private final ConcurrentHashMap<String, Long> lastSendTimeList;
////    /**
////     * 用户一秒内发送的消息数---滑动窗口实现  （不能大于 messageCountPerSec）
////     */
////    private final ConcurrentHashMap<String, Integer> messageRateLimiter;
//    /**
//     * 限流配置：每个用户每秒最多发送5条消息
//     */
//    private static final int MAX_MESSAGES_PER_SECOND = ConfigReader.getInt("Chat.server.message.count.per.second", 5);
//    private static final int RATE_LIMIT_WINDOW_MS = 1000; // 1秒
//    /**
//     * 消息队列 - 缓冲待发送的广播消息
//     */
//    private final BlockingQueue<MessageVO> broadcastQueue = new LinkedBlockingQueue<>(10000);
//
//
//    public ChatServer(String host, int port) {
//        this.host = host;
//        this.port = port;
//        this.userCurMessageList = new ConcurrentHashMap<>();
//        this.chatUserList = new ConcurrentHashMap<>();
//        this.lastSendTimeList = new ConcurrentHashMap<>();
////        this.messageRateLimiter = new ConcurrentHashMap<>();
//        // 启动广播消息处理器
//        startBroadcastProcessor();
//        perSecMsgQueue = new ConcurrentHashMap<>();
//    }
//
//    /**
//     * 启动广播消息处理器，从队列中获取消息并异步发送
//     */
//    private void startBroadcastProcessor() {
//        new Thread(()->{
//            while (true){
//                try {
//                    MessageVO message = broadcastQueue.take();
//                    messageSenderExecutor.submit(new Runnable() {
//                        @Override
//                        public void run() {
//                            sendToAllClients(message);
//                        }
//                    });
//                }catch (InterruptedException e){
//                    log.debug("ChatServer startBroadcastProcessor 广播处理器被中断");
//                    Thread.currentThread().interrupt();
//                    break;
//                }catch (Exception e){
//                    log.debug("ChatServer startBroadcastProcessor 异常",e);
//                }
//            }
//        },"broadcast-processor").start();
//    }
//
//    private void sendToAllClients(MessageVO message) {
//        Long senderKey = message.getSenderId();
//        Set<Map.Entry<Long, Socket>> entries = chatUserList.entrySet();
//
//        for (Map.Entry<Long, Socket> entry : entries) {
//            // TODO 如果message是广播消息，不发送给自己  同时  还要发送同会话的群成员  同时还要分 已读、未读情况
//
//            if(ConversationType.GROUP.getCode().equals(message.getSessionVO().getConversationType())
//                    && entry.getKey().equals(senderKey)){
//                // 是群聊，并且遍历到了自己  不发送消息
//                continue;
//            }
//            // 发送消息
//            Socket receiver = entry.getValue();
//            if (receiver != null && !receiver.isClosed()) {
//                MessageVO messageTemp = new MessageVO();
//                BeanUtils.copyProperties(message,messageTemp);
//                messageTemp.setReceiverId(entry.getKey());
//
//
//                messageSenderExecutor.submit(new Runnable() {
//                    @Override
//                    public void run() {
//                        sendToSingleClient(userCurMessageList,receiver, messageTemp);
//                    }
//                });
//            }
//        }
//    }
//
//    /**
//     * sendToSingleClient 必须是静态方法  因为sendToSingleClient写在了主线程，子线程跟主线程之间的通信就会出现问题
//     *
//     * @param userCurMessageList
//     * @param message
//     */
//    public static void sendToSingleClient(ConcurrentHashMap<Long, CopyOnWriteArrayList<MessageVO>> userCurMessageList,
//                                          Socket receiver,
//                                          MessageVO message) {
//        Long senderKey = message.getSenderId();
//        // 设置当前用户发送的消息列表
//        if (userCurMessageList.containsKey(senderKey)) {
//            CopyOnWriteArrayList<MessageVO> userMsgList = userCurMessageList.get(senderKey);
//            userMsgList.add(message);
//        }else {
//            CopyOnWriteArrayList<MessageVO> messages = new CopyOnWriteArrayList<>();
//            messages.add(message);
//            userCurMessageList.put(senderKey,messages);
//        }
////        Socket receiver = chatUserList.get(senderKey);
//        if (receiver == null || receiver.isClosed()) {
//            log.warn("Socket is already closed, cannot send message.");
//        }
//    }
//
////    private static void removeDisconnectedClient(Socket receiver) {
////        chatUserList.entrySet().removeIf(entry -> {
////            try {
////                // TODO 保存用户的聊天缓存记录到数据库中
////                return entry.getValue().isClosed() ||
////                        entry.getValue().getInputStream().available() < 0;
////            } catch (IOException e) {
////                return true;
////            }
////        });
////    }
//
//    public void startServer() {
//        ServerSocket serverSocket = null;
//        try {
//            serverSocket = new ServerSocket(port);
//            log.info("服务器已启动，监听端口: {}", port);
//
//            // CacheConstants.CHAT_SERVER_SOCKET_KEY + IpUtils.getLocalIp()
//            redisCache.setCacheObject(CacheConstants.CHAT_SERVER_SOCKET_KEY + IpUtils.getLocalIp(), serverSocket);
//
//            while (true){
//                Socket clientSocket = serverSocket.accept();
//                log.info("新客户端连接: " + clientSocket.getInetAddress().getHostAddress());
//                clientHandlerExecutor.submit(new ClientHandler(clientSocket));
//            }
//
//        }catch (IOException e){
//            log.debug("服务器启动失败",e);
//        }finally {
//            try {
//                if (serverSocket != null) {
//                    serverSocket.close();
//                }
//                clientHandlerExecutor.shutdown();
//                messageSenderExecutor.shutdown();
//                saveMsgExecutor.shutdown();
//            } catch (IOException e) {
//                log.error("关闭服务器失败", e);
//            }
//        }
//    }
//
//
//
//    @Data
//    @AllArgsConstructor
//    @NoArgsConstructor
//    private static class BroadcastMessage {
//        private String senderKey;
//        private String receiveKey;
//        private String content;
//        private Date sendTime;
//        private boolean isBroadcast;
//    }
//
//    private class ClientHandler implements Runnable {
//
//        private final Socket clientSocket;
//        private Long clientKey;
//        private BufferedReader in;
//        private PrintWriter out;
//        public ClientHandler(Socket clientSocket) {
//            this.clientSocket = clientSocket;
//        }
//
//        @Override
//        public void run() {
//
//            try {
//                in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
//                out = new PrintWriter(clientSocket.getOutputStream(), true);
//
//                // 接受客户端基本信息
//                String clientId = in.readLine();
//                String loginUserJson = in.readLine();
//                if (clientId == null || loginUserJson == null) {
//                    log.warn("客户端信息不完整，断开连接");
//                    return;
//                }
//
//                // TODO 设置用户连接到服务器的socket
////                redisCache.setCacheMapValue(clientId,loginUserJson,clientSocket);
//
//                clientKey = Long.parseLong(clientId);
//                chatUserList.put(clientKey, clientSocket);
//                log.info("ChatServer 新用户加入: "+clientKey);
////                Set<Map.Entry<String, Socket>> entries = chatUserList.entrySet();
////                for (Map.Entry<String, Socket> entry : entries) {
////                    PrintWriter reOut = null;
////                    Socket socket = chatUserList.get(entry.getKey());
////                    if(socket != null){
////                        reOut = new PrintWriter(socket.getOutputStream(), true);
////                        if(entry.getKey().contains(clientKey)){
////                            reOut.println("欢迎加入聊天室，当前在线人数: " + chatUserList.size());
////                        }else {
////                            reOut.println("新用户加入: "+clientKey +" 当前在线人数: " + chatUserList.size());
////                        }
////                    }
////                }
//
//                //处理消息
//                String inputLine;
//                while ((inputLine = in.readLine()) != null) {
//                    log.info("收到 {} 的消息: {}", clientKey, inputLine);
//                    if ("exit".equalsIgnoreCase(inputLine)) {
//                        break;
//                    }
//                    // 检查当前用户消息发送频率是否超过限制
//                    // 限制1.一秒内发送不得超过5条消息
//                    if (isRateLimited(clientKey)) {
//                        log.warn("用户 {} 发送消息频率过高，被限制", clientKey);
//                        continue;
//                    }
//                    MessageVO messageVO = JSON.parseObject(inputLine, MessageVO.class);
//                    if(ObjectUtils.isEmpty(messageVO) || ObjectUtils.isEmpty(messageVO.getSessionVO())){
//                        log.warn("用户 {} 发送消息格式错误 {}",clientKey, messageVO);
//                        continue;
//                    }
//
//                    if (ConversationType.GROUP.getCode().equals(messageVO.getSessionVO().getConversationType())) {
////                        String[] split = inputLine.split(" ");
////                        String content = split.length > 1 ? split[1] : "";
////                        String broadcastContent = clientName + " 发送广播消息，给您发送的内容是: " + content;
////
////                        if(!broadcastQueue.offer(new BroadcastMessage(clientKey, null, broadcastContent, new Date(), true))){
////                            out.println("消息发送失败，服务器繁忙");
////                            log.warn("广播队列已满，消息丢弃");
////                        }else {
////                            out.println(clientName + "的广播消息已发送");
////                        }
//                    }
//                    else if(ConversationType.SINGLE.getCode().equals(messageVO.getSessionVO().getConversationType())){
//                        // 一对一聊天
//                        Long receiverId = messageVO.getReceiverId();
//                        String sessionId = messageVO.getSessionVO().getId();
//                        String receiverExistFlag = redisCache.getCacheObject(CacheConstants.CHAT_USER_EXIST_FLAG + receiverId);
//
//                        if("1".equals(receiverExistFlag)){
//                            /**
//                             * 接口1： 点击会话列表，设置已进入会话列表，使用map存储
//                             * 存在逻辑
//                             *      在线，同时用户点击了聊天会话  才能将消息设置为已读
//                             */
//                            SysUserVO receiver = messageVO.getReceiver();
//                            if(ObjectUtils.isNotEmpty(receiver)){
//
//                                Map<String, Boolean> entrySessionMap = redisCache.getCacheMap(CacheConstants.CHAT_USER_SESSION + SecurityUtils.getUserId());
//
//                                if(ObjectUtils.isNotEmpty(entrySessionMap) && entrySessionMap.containsKey(sessionId)){
//                                    boolean containsKey = entrySessionMap.containsKey(sessionId);
//                                    handleMessageInfo(messageVO,containsKey,false);
//                                }
//
//                            }else {
//                                log.info("消息格式错误，messageVO.receiver is null");
//                            }
//                        }
//                        else {
//                            /**
//                             * 不存在逻辑
//                             *      继续发送消息，但是设置为未读状态
//                             */
//                            handleMessageInfo(messageVO,false, false);
//                        }
//                    }
//                }
//            }catch (IOException e){
//                log.error("处理客户端 {} 时出错", clientKey, e);
//            }finally {
//                // 清理资源
//                log.info("cleanup 被触发");
//                cleanup(clientKey);
//            }
//        }
//
//        private void handleMessageInfo(MessageVO messageVO, Boolean isRead, boolean isGroup) {
//            messageVO.setIsRead(isRead?MessageIsRead.YES.getCode():MessageIsRead.NO.getCode());
//            if (messageVO.getMessageType() != null) {
//
//                switch (messageVO.getMessageType()){
//                    case 1:
//                        // 文字
//                        if(isGroup){
//                            sendToAllClients(messageVO);
//                        }else {
//                            sendPrivateMessage(messageVO);
//                        }
//                        break;
//                    case 2:
//                        // 图片
//                        break;
//                    case 3:
//                        // 文件
//                        break;
//                    case 4:
//                        // 引用消息
//
//                        if (StringUtils.isNotEmpty(messageVO.getQuoteMessageId())) {
//
//
//                        }
//
//                        break;
//                    case 5:
//                        // 转发消息
//                        if(StringUtils.isNotEmpty(messageVO.getForwardFromId())){
//
//                        }
//                        break;
//                    default:
//                        log.info("未知的消息类型");
//                }
//            }
//        }
//
//        private void cleanup(Long clientKey) {
//            try {
//                if(in != null){
//                    in.close();
//                }
//                if(out != null){
//                    out.close();
//                }
//                if(clientSocket != null && !clientSocket.isClosed()){
//                    clientSocket.close();
//                }
//                if(clientKey != null){
//                    redisCache.setCacheObject(CacheConstants.CHAT_USER_EXIST_FLAG + clientKey, "0");
//
//                    chatUserList.remove(clientKey);
//
//                    // TODO 异步保存用户的聊天缓存记录到数据库中---位置待定
//                    CopyOnWriteArrayList<MessageVO> broadcastMessages = userCurMessageList.get(clientKey);
//                    saveMsgExecutor.submit(() -> saveMsgList(broadcastMessages));
//
//                    userCurMessageList.remove(clientKey);
//
//                    perSecMsgQueue.remove(clientKey);
//                    log.info("用户 {} 已退出，当前在线人数: {}", clientKey, chatUserList.size());
//                }
//            }catch (IOException e){
//                log.error("清理客户端资源失败",e);
//            }
//        }
//
//        private void saveMsgList(CopyOnWriteArrayList<MessageVO> broadcastMessages) {
//
//            List<MessageVO> collect = broadcastMessages.stream().collect(Collectors.toList());
//
//            //保存，这里还没有整合项目，就做打印操作   xxx.saveBath(List<BroadcastMessage> collect);
//            for (MessageVO messageVO : collect) {
//                messageService.insertMessage(messageVO);
//            }
//        }
//
//        /**
//         *
//         * @param messageVO
//         */
//        private void sendPrivateMessage(MessageVO messageVO) {
//            Long senderId = messageVO.getSenderId();
//            Long receiverId = messageVO.getReceiverId();
//            String sessionId = messageVO.getSessionId();
//            Socket receiverSocket = chatUserList.get(receiverId);
//            if(receiverSocket == null){
//                log.warn(receiverId + "未上线");
//            }else {
//
//                List<MessageVO> messageVOList = redisCache.getCacheMapValue(CacheConstants.CHAT_USER_SESSION + senderId, sessionId);
//                if(CollectionUtils.isEmpty(messageVOList)){
//                    messageVOList = new ArrayList<>();
//                    messageVOList.add(messageVO);
//                }else {
//                    messageVOList.add(messageVO);
//                }
//                redisCache.setCacheMapValue(CacheConstants.CHAT_USER_SESSION + senderId, messageVO.getSessionId(), messageVOList);
//
//                111
////                messageSenderExecutor.submit(() -> sendToSingleClient(userCurMessageList, receiverSocket, messageVO));
////                sendToSingleClient(receiverSocket, message);
//            }
//        }
//
//
//        private boolean isRateLimited(Long clientKey) {
//            return getCount(clientKey) > MAX_MESSAGES_PER_SECOND;
//        }
//
//
//        /**
//         * 消息队列
//         *  添加数据并自动清理过期元素
//         */
//        public void add(MessageVO message,BlockingQueue<MessageVO> dataList) {
//            long currentTime = System.currentTimeMillis();
//
//            // 先清理窗口外的过期数据
//            removeExpiredData(currentTime,dataList);
//
//            // 添加新数据到队列尾部
//            dataList.add(message);
//        }
//
//        // 清理过期数据
//        private void removeExpiredData(long currentTime, BlockingQueue<MessageVO> dataList) {
//            long threshold = currentTime - RATE_LIMIT_WINDOW_MS;
//
//            // 从队头开始移除所有过期数据
//            while (!dataList.isEmpty()) {
//                MessageVO first = null;
//                try {
//                    first = dataList.take();
//                    // 添加发送时间属性
//                    if (ObjectUtils.isNotEmpty(first.getSendTime()) && first.getSendTime().getTime() < threshold) {
//                        boolean remove = dataList.remove(first);
//                        if (!remove) {
//                            log.warn("ClientHandler removeExpiredData 数据移除失败");
//                        }
//                    } else {
//                        // 队头元素未过期，后续遍历的元素也不会过期（队列按时间有序）
//                        break;
//                    }
//                } catch (InterruptedException e) {
//                    log.warn("ClientHandler removeExpiredData 出错：{}",e.getMessage());
//                }
//            }
//        }
//
//        // 获取当前窗口内的元素数量
//        public int getCount(Long clientKey) {
//            // 先清理可能的过期数据
//            BlockingQueue<MessageVO> messages = perSecMsgQueue.get(clientKey);
//            if(messages == null){
//                return 0;
//            }
//            removeExpiredData(System.currentTimeMillis(),messages);
//            return messages.size();
//        }
//
//        // 获取窗口内的所有元素
//        public BlockingQueue<MessageVO> getElements(String clientKey) {
//            BlockingQueue<MessageVO> messages = perSecMsgQueue.get(clientKey);
//            if(messages == null){
//                messages = new LinkedBlockingQueue<>();
//            }
//            removeExpiredData(System.currentTimeMillis(),messages);
//            return new LinkedBlockingQueue<>(messages); // 返回副本防止外部修改
//        }
//    }
//
//}
